package com.intct.ods;

import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @author gufg
 * @since 2025-07-02 15:38
 */
public class OdsSQLApplication {
    public static void main(String[] args) throws Exception {
        // 配置属性
        Configuration conf = new Configuration();
        conf.set(RestOptions.BIND_PORT, "8081");

        // 创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.days(30),Time.seconds(3)));

        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
//        tEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(30 * 60 + 5));

        tEnv.executeSql("CREATE TABLE area_info (" +
                "db_name STRING METADATA FROM 'database_name' VIRTUAL," +
                "table_name STRING METADATA  FROM 'table_name' VIRTUAL," +
                "operation_ts TIMESTAMP_LTZ(3) METADATA FROM 'op_ts' VIRTUAL," +
                " code INT," +
                " name STRING," +
                " PRIMARY KEY (code) NOT ENFORCED" +
                ") WITH (" +
                " 'connector' = 'mysql-cdc'," +
                " 'hostname' = 'cdh-node'," +
                " 'port' = '13306'," +
                " 'username' = 'root'," +
                " 'password' = 'Test_090110'," +
                " 'database-name' = '^(travel).*'," +
                " 'table-name' = '.*'," +
                " 'server-time-zone' = 'Asia/Shanghai'," +
                " 'scan.startup.mode' = 'earliest-offset'" +
                ")");

        tEnv.sqlQuery("select * from area_info").execute().print();

        // 创建Sink Kafka表
//        tEnv.executeSql("CREATE TABLE area_info_sink (" +
//            " code INT," +
//            " name STRING," +
//            " PRIMARY KEY (code) NOT ENFORCED" +
//            ") WITH (" +
//            " 'connector' = 'upsert-kafka'," +
//            " 'topic' = 'ods_area_info'," +
//            " 'properties.bootstrap.servers' = 'cdh-node:9092'," +
//            " 'key.format' = 'json'," +
//            " 'value.format' = 'json'" +
//            ")");
//
//        tEnv.executeSql("insert into area_info_sink select * from area_info");
    }
}
